package com.guchenbo.spark.sql


import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoTable, LogicalPlan}
import org.apache.spark.sql.execution.SparkSqlParser
import org.apache.spark.sql.internal.SQLConf

/**
 * Spark SQL 输入输出表解析
 */
object SparkSqlParserDemo {

  def main(args: Array[String]): Unit = {
    // 查询SQL
    var sql = "insert overwrite table dwd.user_info\n" + "select * from ods.user_info\n" + "union all\n" + "select * from ods.user_info_inc"

    sql = "SELECT m.name AS model_name,\n" + "       m.owner,\n" + "       b.name AS dept_name,\n" + "       m.gmt_create,\n" + "       mc.name AS category_name,\n" + "       t3.widget_name,\n" + "       t3.widget_id,\n" + "       t3.board_id,\n" + "       t3.open as alarm_on,\n" + "       t3.board_name,\n" + "       t3.alarm_name,\n" + "       t3.alarm_data,\n" + "       t3.alarm_type,\n" + "       t3.receivers,\n" + "       t3.statis_time,\n" + "       t3.alarm_content,\n" + "       t3.alarm_time,\n" + "       t3.alarm_id,\n" + "       t3.board_create,\n" + "       CASE m.stage\n" + "           WHEN 0 THEN '需求'\n" + "           WHEN 1 THEN '开发'\n" + "           WHEN 2 THEN '验证'\n" + "           WHEN 3 THEN '投产'\n" + "           WHEN 4 THEN '使用中'\n" + "           WHEN 5 THEN '已变更'\n" + "           WHEN 6 THEN '已退役'\n" + "           WHEN 7 THEN '审批通过已完结'\n" + "           WHEN 8 THEN '审批拒绝已完结' END AS state\n" + "FROM model_info m\n" + "         LEFT JOIN bifrost_org b ON m.org_code = b.uuid\n" + "         LEFT JOIN model_category mc ON m.category_code = mc.uuid\n" + "         LEFT JOIN\n" + "\t\t        (SELECT \n" + "                 mm.model_id,\n" + "                 mm.report_name as widget_name,\n" + "                 mm.board_id,\n" + "                 mm.create_time as board_create,\n" + "                 t2.widget_id,\n" + "                 t2.open,\n" + "                 mt.template_name as board_name,\n" + "                 t2.alarm_name,\n" + "                 t2.alarm_data,\n" + "                 t2.alarm_type,\n" + "                 t2.receivers,\n" + "                 t2.statis_time,\n" + "                 t2.alarm_content,\n" + "                 t2.alarm_time,\n" + "                 t2.alarm_id\n" + "\t\t        FROM monitor mm\n" + "\t\t\t         left join \n" + "\t\t\t\t          (select ap.widget_id,\n" + "                          ap.open,\n" + "                          ap.alarm_type,\n" + "                          ap.receivers,\n" + "                          ap.statis_time,\n" + "                          ar.name as alarm_name,\n" + "                          ar.alarm_data,\n" + "                          ar.alarm_content,\n" + "                       \t  ar.alarm_time,\n" + "                       \t  ar.alarm_id,\n" + "                          bk.widget_name,\n" + "                       \t  bk.board_id as a_bid\n" + "                \t from bi_widget_alarm ap\n" + "                     left join bi_widget_alarm_value ar \n" + "                        on ap.widget_id = ar.widget_id\n" + "                     left join bi_dashboard_widgets bk \n" + "                        on ap.widget_id = bk.id) t2\n" + "      \t\t\t     on cast(mm.board_id as decimal) = cast(t2.a_bid as decimal)\n" + "                left join monitor_template mt ON mm.template_id = mt.id\n" + "\t\t        WHERE mm.monitor_type = 0) t3\n" + "\t      ON m.uuid = t3.model_id\n" + "WHERE m.version_type = 1\n" + "    AND (m.is_delete = 0 OR m.is_delete IS NULL)"

//    sql = "select name from table"
//    sql = "select  * from table"
    // 解析 SQL，获得 LogicalPlan
    val sparkSqlParser = new SparkSqlParser(new SQLConf)
    val logicalPlan: LogicalPlan = sparkSqlParser.parsePlan(sql)
    println(logicalPlan)

//    println(logicalPlan.output)

    // 遍历 LogicalPlan，查找输入输出表
//    findTables(logicalPlan);

//    println(sparkSqlParser.parseTableIdentifier(sql))
  }

  /**
   * 遍历 LogicalPlan，查找输入输出表
   *
   * @param logicalPlan 逻辑计划
   */
  def findTables(logicalPlan: LogicalPlan): Unit = {
    println(logicalPlan)
    if (logicalPlan == null)
      return;

    logicalPlan match {
      case logicalPlan: InsertIntoTable => // insert into xxx.xxx、insert overwrite table xxx.xxx
        val insertIntoTable: InsertIntoTable = logicalPlan.asInstanceOf[InsertIntoTable]
        val tableIdentifier: TableIdentifier = insertIntoTable.table.asInstanceOf[UnresolvedRelation].tableIdentifier
        val tableName = tableIdentifier.database.get + "." + tableIdentifier.table
        println("输出表：" + tableName)

      case logicalPlan: UnresolvedRelation => // from xxx.xxx
        val unresolvedRelation: UnresolvedRelation = logicalPlan.asInstanceOf[UnresolvedRelation]
        val tableIdentifier: TableIdentifier = unresolvedRelation.tableIdentifier
        val tableName = tableIdentifier.database.get + "." + tableIdentifier.table
        println("输入表：" + tableName)

      case _ =>
        println
    }

    // 遍历子节点
    for (child <- logicalPlan.children) {
      findTables(child);
    }
  }
}


